package com.zhang.sql2;

import com.zhang.bean.Bean1;
import com.zhang.bean.Bean2;
import com.zhang.util.MysqlUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @title:
 * @author: zhang
 * @date: 2022/6/19 15:14
 */
public class FinkSQLLeftJoinSInkMysql {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        System.out.println(tableEnv.getConfig().getIdleStateRetention());
        //设置状态10s
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        SingleOutputStreamOperator<Bean1> leftStream = env
                .socketTextStream("hadoop102", 9999)
                .map(new MapFunction<String, Bean1>() {
                    @Override
                    public Bean1 map(String value) throws Exception {
                        String[] strings = value.split(" ");
                        return new Bean1(
                                strings[0],
                                strings[1],
                                Long.parseLong(strings[2])
                        );
                    }
                });

        SingleOutputStreamOperator<Bean2> rightStream = env
                .socketTextStream("hadoop102", 8888)
                .map(new MapFunction<String, Bean2>() {
                    @Override
                    public Bean2 map(String value) throws Exception {
                        String[] strings = value.split(" ");
                        return new Bean2(
                                strings[0],
                                strings[1],
                                Long.parseLong(strings[2])
                        );
                    }
                });


        //将流转换为动态表
        tableEnv.createTemporaryView("t1", leftStream);
        tableEnv.createTemporaryView("t2", rightStream);

        Table table = tableEnv.sqlQuery("select t1.id,t1.name,t2.sex from t1 left join t2 on t1.id = t2.id");
        tableEnv.createTemporaryView("t", table);
        //创建sink表
        //创建Kafka表
        tableEnv.executeSql("" +
                "create table result_table(" +
                "    id string," +
                "    name string," +
                "    sex string," +
                "    PRIMARY KEY (id) NOT ENFORCED " +
                ") " + MysqlUtil.mysqlTableDDL("test"));

        //将数据写入Mysql
        tableEnv.executeSql("insert into result_table select * from t")
                .print();


        env.execute();
    }
}
